package io.kiki.sba.registry.compress;

import io.kiki.sba.registry.cache.Sizer;
import io.kiki.sba.registry.metrics.CounterFunc;
import io.kiki.sba.registry.util.StringFormatter;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

public final class CompressUtils {
    private static final Logger LOG = LoggerFactory.getLogger(CompressUtils.class);
    private static final Map<String, Compressor> compressorMap = new HashMap<String, Compressor>() {
        {
            final Compressor gzipCompressor = new Compressor.GzipCompressor();
            final Compressor zstdCompressor = new Compressor.ZstdCompressor();

            put(gzipCompressor.getEncoding(), gzipCompressor);
            put(zstdCompressor.getEncoding(), zstdCompressor);
        }
    };

    private CompressUtils() {
    }

    public static Compressor mustGet(String encode) {
        Compressor compressor = compressorMap.get(encode);
        Assert.notNull(compressor, StringFormatter.format("compress {} not found", encode));
        return compressor;
    }

    public static Compressor find(String[] acceptEncodes) {
        return find(acceptEncodes, Collections.emptySet());
    }

    public static Compressor find(String[] acceptEncodes, Set<String> forbidEncodes) {
        if (ArrayUtils.isEmpty(acceptEncodes)) {
            return null;
        }
        for (String encoding : acceptEncodes) {
            if (forbidEncodes.contains(encoding)) {
                continue;
            }
            Compressor compressor = compressorMap.get(encoding);
            if (compressor != null) {
                return compressor;
            }
        }
        LOG.warn("accept encoding {} not in available compressors", StringUtils.join(acceptEncodes, ","));
        return null;
    }

    public static <V extends Sizer> CompressCachedExecutor<V> newCachedExecutor(String name, long silentMs, long maxWeight) {
        CompressCachedExecutor<V> cachedExecutor = new CompressCachedExecutor<>(name, silentMs, maxWeight);
        CounterFunc cacheCounter = CounterFunc.build().namespace("compress").subsystem("cache").name(name).labelNames("type").help(StringFormatter.format("compress cache {} hit or missing", name)).create().register();
        cacheCounter.labels("hit").func(cachedExecutor::getHitCount);
        cacheCounter.labels("missing").func(cachedExecutor::getMissingCount);
        return cachedExecutor;
    }

    public static String normalizeEncode(String encode) {
        if (StringUtils.isBlank(encode)) {
            return "plain";
        }
        return encode;
    }
}
